package SparkStreamingKafka

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming {

  def main(args: Array[String]) {
    //zookeeper的地址
    val zkQuorum = "192.168.11.31:2181/home/hadoop/app/kafka_2.12-2.2.0"
    //group_id可以通过kafka的conf下的consumer.properties中查找
    val group = "test-consumer-group"
    //创建的topic 可以是一个或多个
    val topics = "SparkKafka"
    val sparkConf = new SparkConf().setMaster("local").setAppName("SparkStreaming").set("spark.executor.memory", "1g")
    val sc = new StreamingContext(sparkConf, Seconds(5))
    val numThreads = 2
    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
    //StorageLevel.MEMORY_AND_DISK_SER为存储的级别
    val lines = KafkaUtils.createStream(sc, zkQuorum, group, topicMap, StorageLevel.MEMORY_AND_DISK_SER).map(_._2)
    //对于收到的消息进行wordcount
    val words = lines.flatMap(_.split(" "))
    val pairs = words.map(word => (word, 1))
    val wordCounts = pairs.reduceByKey(_ + _)
    wordCounts.print()
    sc.start()
    sc.awaitTermination()
  }

}
